package com.atguigu.flink.sql.function;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * Created by Smexy on 2023/4/14
 *
 *  表值函数 将标量值转换成新的行数据；
 *          1行N列进，N行N列出的函数。
 *          表生成函数，类似炸裂。
 *
 *   输入: sensor_1_a_b
 *   输出:   拆分的结果     输出字符的长度
 *           sensor         14
 *            1              14
 *            a             14
 *            b              14
 *
 *   ①按照flink的规范，定义一个函数类
 *   ②在系统中注册函数，提供一个函数名字
 *   ③可以调用函数
 */
public class Demo5_CustomUDTFFunciton
{
    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        String createTableSql = " create table t1 ( id STRING, ts BIGINT , vc INT " +
            "    )WITH (" +
            "  'connector' = 'filesystem'," +
            "  'path' = 'data/sensor.txt'," +
            "  'format' = 'csv'" +
            ") ";

        tableEnvironment.executeSql(createTableSql);

        //②在系统中注册函数，提供一个函数名字
        MyExplode myFunction = new MyExplode();
        tableEnvironment.createTemporaryFunction("myExplode",myFunction);


        /*
            udtf 在hive中 需要使用 lateral view 语法声明。
            laterval view语法的底层原理是 join!

            表生成函数，需要把函数生成的结果表和之前的表进行join，返回最终的结果
                inner join： 内连接，返回交集
                left join : 左外连接，返回左侧的全部
         */
        tableEnvironment.sqlQuery(" select * from t1 " +
            //"                       inner JOIN LATERAL TABLE(myExplode(id)) ON TRUE" )
           "                       LEFT JOIN LATERAL TABLE(myExplode(id)) ON TRUE" )
                        .execute().print();

    }

    /*
        你需要扩展 org.apache.flink.table.functions 下的 TableFunction

              通过 eval 方法计算，没有返回值。通过 collect(T) 方法来发送要输出的行。
              如果一行有多列，如何封装T：
                    选择1： 自定义POJO
                    选择2:  省事，使用Row

           @FunctionHint： 函数提示。提示输出的每一行的元数据信息。
     */

    @FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
    public static class MyExplode extends TableFunction<Row>
    {
        public void eval(String id){

            //当前行的输入是 sensor_3，不输出
            if (!"sensor_3".equals(id)){
                String[] words = id.split("_");

                for (String word : words) {
                    //输出一行
                    collect(Row.of(word,id.length()));
                }
            }
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc{

        private Integer num;
        private Double sumVc;
     }
}
